package pl.com.qapps.datastore;

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.google.appengine.api.datastore.AsyncDatastoreService;
import com.google.appengine.api.datastore.DatastoreAttributes;
import com.google.appengine.api.datastore.DatastoreServiceConfig;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Index;
import com.google.appengine.api.datastore.Index.IndexState;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyRange;
import com.google.appengine.api.datastore.PreparedQuery;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.api.datastore.TransactionOptions;
import com.google.appengine.api.utils.FutureWrapper;

public class AbstractAsyncDatastoreService implements AsyncDatastoreService {

	protected final DatastoreServiceConfig config;
	protected final ExecutorService executor;
	protected final TransactionStack txnStack;
	protected final Store store;

	public AbstractAsyncDatastoreService(DatastoreServiceConfig config,
			ExecutorService executor, TransactionStack txnStack, Store store) {
		this.config = Objects.requireNonNull(config);
		this.executor = Objects.requireNonNull(executor);
		this.txnStack = Objects.requireNonNull(txnStack);
		this.store = Objects.requireNonNull(store);
	}

	@Override
	public Collection<Transaction> getActiveTransactions() {
		return txnStack.getAll();
	}

	@Override
	public Transaction getCurrentTransaction() {
		return txnStack.peek();
	}

	@Override
	public Transaction getCurrentTransaction(Transaction returnedIfNoTxn) {
		return txnStack.peek(returnedIfNoTxn);
	}

	@Override
	public PreparedQuery prepare(Query query) {
		if (query == null)
			throw new NullPointerException("query cannot be null");
		// TODO implement
		throw new UnsupportedOperationException("not supported yet");
	}

	@Override
	public PreparedQuery prepare(Transaction txn, Query query) {
		if (query == null)
			throw new NullPointerException("query cannot be null");
		throw new UnsupportedOperationException("not supported yet");
		// if (txn != null) {
		// if (query.getAncestor() == null)
		// throw new IllegalArgumentException(
		// "txn is not null and query is not ancestor query");
		// verifyTxn(txn);
		// return store.newAncestorQuery(txn, query);
		// } else
		// return store.newQuery(query);
	}

	@Override
	public Future<KeyRange> allocateIds(String kind, long num) {
		return allocateIds(null, kind, num);
	}

	@Override
	public Future<KeyRange> allocateIds(Key parent, String kind, long num) {
		// TODO Auto-generated method stub
		throw new UnsupportedOperationException("not supported yet");
	}

	@Override
	public Future<Transaction> beginTransaction() {
		return beginTransaction(TransactionOptions.Builder.withDefaults());
	}

	@Override
	public Future<Transaction> beginTransaction(final TransactionOptions options) {
		if (options == null)
			return beginTransaction(TransactionOptions.Builder.withDefaults());
		// TODO find better way to register transaction in caller thread than
		// FutureWrapper that registers transaction in Future.get() variants
		// callers but not this method callers.
		return new FutureWrapper<Transaction, Transaction>(
				executor.submit(new Callable<Transaction>() {

					@Override
					public Transaction call() throws Exception {
						// TODO implement
						throw new UnsupportedOperationException(
								"not supported yet");
						// return store.newTransaction(options);
					}
				})) {

			@Override
			protected Throwable convertException(Throwable e) {
				return e;
			}

			@Override
			protected Transaction wrap(Transaction txn) throws Exception {
				return txn;
			}

			@Override
			public Transaction get() throws ExecutionException,
					InterruptedException {
				return push(super.get());
			}

			@Override
			public Transaction get(long timeout, TimeUnit unit)
					throws InterruptedException, TimeoutException,
					ExecutionException {
				return push(super.get());
			}

			private Transaction push(Transaction txn) {
				Transaction current = txnStack.peek();
				if (current == null || !txn.getId().equals(current.getId()))
					;
				txnStack.push(txn);
				return txn;
			}

		};
	}

	@Override
	public Future<Void> delete(Key... keys) {
		return delete(null, keys);
	}

	@Override
	public Future<Void> delete(Iterable<Key> keys) {
		return delete(null, keys);
	}

	@Override
	public Future<Void> delete(Transaction txn, Key... keys) {
		return delete(txn, Arrays.asList(keys));
	}

	@Override
	public Future<Void> delete(final Transaction transaction,
			final Iterable<Key> keys) {
		return executor.submit(new Callable<Void>() {

			@Override
			public Void call() throws Exception {
				verifyKeys(keys);
				// TODO implement
				throw new UnsupportedOperationException("not supported yet");
				// store.makeDelete(verifyTxn(transaction), keys);
				// return null;
			}
		});
	}

	@Override
	public Future<Entity> get(Key key) {
		return get(null, key);
	}

	@Override
	public Future<Map<Key, Entity>> get(Iterable<Key> keys) {
		return get(null, keys);
	}

	@Override
	public Future<Entity> get(Transaction txn, final Key key) {
		return new FutureWrapper<Map<Key, Entity>, Entity>(get(txn,
				Arrays.asList(key))) {

			@Override
			protected Throwable convertException(Throwable wrapped) {
				return wrapped;
			}

			@Override
			protected Entity wrap(Map<Key, Entity> wrapped) throws Exception {
				return wrapped.get(key);
			}

		};
	}

	@Override
	public Future<Map<Key, Entity>> get(final Transaction transaction,
			final Iterable<Key> keys) {
		return executor.submit(new Callable<Map<Key, Entity>>() {

			@Override
			public Map<Key, Entity> call() throws Exception {
				verifyKeys(keys);
				// TODO implement
				throw new UnsupportedOperationException("not supported yet");
				// final Transaction txn = verifyTxn(transaction);
				// if (transaction != null)
				// return store.makeTransactionalGet(txn, keys);
				// else
				// return store.makeGet(keys);
			}

		});
	}

	@Override
	public Future<DatastoreAttributes> getDatastoreAttributes() {
		// TODO Auto-generated method stub
		throw new UnsupportedOperationException(" not supported yet");
	}

	@Override
	public Future<Map<Index, IndexState>> getIndexes() {
		// TODO Auto-generated method stub
		throw new UnsupportedOperationException(" not supported yet");
	}

	@Override
	public Future<Key> put(Entity entity) {
		return put(null, entity);
	}

	@Override
	public Future<List<Key>> put(Iterable<Entity> entities) {
		return put(null, entities);
	}

	@Override
	public Future<Key> put(Transaction txn, Entity entity) {
		return new FutureWrapper<List<Key>, Key>(
				put(txn, Arrays.asList(entity))) {

			@Override
			protected Throwable convertException(Throwable wrapped) {
				return wrapped;
			}

			@Override
			protected Key wrap(List<Key> keys) throws Exception {
				return keys.get(0);
			}

		};
	}

	@Override
	public Future<List<Key>> put(final Transaction transaction,
			final Iterable<Entity> entities) {
		return executor.submit(new Callable<List<Key>>() {

			@Override
			public List<Key> call() throws Exception {
				final Transaction txn = verifyTxn(transaction);

				// TODO implement
				throw new UnsupportedOperationException("not supported yet");
				// if (transaction != null)
				// return store.makeTransactionalPut(txn, entities);
				// else
				// return store.makePut(entities);
			}
		});
	}

	/**
	 * Override, and call super method, to do some checks. If you won't call
	 * super method you'll break the automatic current transaction feature.
	 * 
	 * @param txn
	 * @throws IllegalStateException
	 *             if txn is not null and not active
	 */
	protected Transaction verifyTxn(Transaction txn)
			throws IllegalStateException {
		if (txn != null) {
			if (!txn.isActive())
				throw new IllegalStateException("txn is not active");
		} else
			txn = getCurrentTransaction();
		return txn;
	}

	protected static void verifyKeys(Iterable<Key> keys) {
		for (Key key : keys) {
			if (key == null)
				throw new IllegalArgumentException("one of given keys is null");
			if (!key.isComplete()) {
				throw new IllegalArgumentException("key '" + key
						+ "' is incomplete");
			}
		}
	}
}
